/**
 * Copyright © 2022 The TyIotCloud Authors: TangZhenzhi
 */
package org.mqtt.server;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.mqtt.MqttDecoder;
import io.netty.handler.codec.mqtt.MqttEncoder;
import io.netty.handler.codec.mqtt.MqttMessageType;
import io.netty.handler.ssl.SslContext;
import io.netty.handler.ssl.SslContextBuilder;
import io.netty.handler.ssl.SslHandler;
import io.netty.handler.ssl.SslProvider;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;

import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLEngine;
import java.io.InputStream;
import java.security.KeyStore;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;

@Slf4j
public class MqttServer {

    @Getter
    private final static List<MqttMessageType> eventsFromClient = new CopyOnWriteArrayList<>();
    @Getter
    private final static int mqttPort = 8885;

    private static Channel serverChannel;
    private static EventLoopGroup bossGroup;
    private static EventLoopGroup workerGroup;

    public static void main(String[] args) throws Exception {

        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try {
            // Load the server certificate and private key
            InputStream keyInputStream = MqttServer.class.getResourceAsStream("/server.key");
            InputStream certInputStream = MqttServer.class.getResourceAsStream("/server.crt");

            SslContext sslNettyContext = SslContextBuilder.forServer(certInputStream, keyInputStream).sslProvider(SslProvider.JDK).build();

            // Configure the server
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) {
                            ChannelPipeline pipeline = ch.pipeline();
                            SSLEngine sslEngine = sslNettyContext.newEngine(ch.alloc());
                            sslEngine.setUseClientMode(false);
                            pipeline.addLast("ssl", new SslHandler(sslEngine));
                            pipeline.addLast("decoder", new MqttDecoder(65536));
                            pipeline.addLast("encoder", MqttEncoder.INSTANCE);

                            MqttTransportHandler handler = new MqttTransportHandler(eventsFromClient);

                            pipeline.addLast(handler);
                            // Add your other handlers here
                        }
                    });

            // Bind and start to accept incoming connections
            bootstrap.bind(1983).sync().channel().closeFuture().sync();


            log.info("Starting MQTT server...");
            bossGroup = new NioEventLoopGroup();
            workerGroup = new NioEventLoopGroup();
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline pipeline = ch.pipeline();
                            pipeline.addLast("decoder", new MqttDecoder(65536));
                            pipeline.addLast("encoder", MqttEncoder.INSTANCE);

                            MqttTransportHandler handler = new MqttTransportHandler(eventsFromClient);

                            pipeline.addLast(handler);
                            ch.closeFuture().addListener(handler);
                        }
                    })
                    .childOption(ChannelOption.SO_KEEPALIVE, true);

            serverChannel = b.bind(mqttPort).sync().channel();
            log.info("Mqtt transport started!");
        } catch (Exception e) {

        }
    }

//    public void shutdown() throws InterruptedException {
//        log.info("Stopping MQTT transport!");
//        try {
//            serverChannel.close().sync();
//        } finally {
//            workerGroup.shutdownGracefully();
//            bossGroup.shutdownGracefully();
//        }
//        log.info("MQTT transport stopped!");
//    }
}
